package com.hzgj.bcl.soa.cicada.sender;

import com.hzgj.bcl.soa.cicada.AsyncHandler;
import com.hzgj.bcl.soa.cicada.ListenableFuture;
import com.hzgj.bcl.soa.cicada.handle.ResponseFuture;
import com.hzgj.bcl.soa.cicada.pool.ChannelPoolManager;
import com.hzgj.bcl.soa.cicada.pool.Channels;
import com.hzgj.bcl.soa.cicada.timeout.RequestTimeoutTimerTask;
import com.hzgj.bcl.soa.cicada.timeout.TimeoutHolder;
import com.hzgj.bcl.soa.rpc.IpAddress;
import com.hzgj.bcl.soa.rpc.message.Message;
import com.hzgj.bcl.soa.util.Constants;
import com.hzgj.bcl.soa.util.Util;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.Timer;
import io.netty.util.TimerTask;
import io.netty.util.concurrent.Future;
import lombok.Getter;
import lombok.Setter;

/**
 * Created by ye.pan on 2015/6/25.
 */
public class CicadaSender implements Sender {

    private final Logger logger = LoggerFactory.getLogger(getClass());
    private final Timer timer;
    @Getter
    @Setter
    private Status status = Status.INVALIDATE;
    @Getter
    private ChannelPoolManager channelPoolManager;
    @Getter
    private IpAddress ipAddress;

    /**
     * @param timer sharing timer across jvm
     */
    public CicadaSender(HashedWheelTimer timer, IpAddress ipAddress) {
        this.timer = timer;
        this.ipAddress = ipAddress;
    }

    public CicadaSender channelPoolManager(ChannelPoolManager channelPoolManager) {
        this.channelPoolManager = channelPoolManager;
        return this;
    }

    @Override
    public void start() {
        status = Status.OK;
    }

    @Override
    public void stop() {
        try {
            timer.stop();
            channelPoolManager.stop();
            status = Status.CLOSED;
        } catch (Throwable t) {
            logger.warn("Unexpected error on closeChannel", t);
        }
    }

    public boolean isOk() {
        return getStatus().equals(Status.OK);
    }

    @Override
    public <T> ListenableFuture<T> send(final Message message, final AsyncHandler<T> asyncHandler, long requestTimeoutInMs, String key) throws IOException {
        if (status == Status.INVALIDATE) {
            throw new IOException("sender invalidate.");
        }

        ResponseFuture<T> future = new ResponseFuture<>(asyncHandler);
        future.setPoolKey(key);

        Future<Channel> channelFuture = channelPoolManager.getChannelPool(key).acquire();
        if (channelFuture.awaitUninterruptibly(2, TimeUnit.SECONDS)) {
            if (!channelFuture.isSuccess()) {
                throw new IOException(channelFuture.cause());
            }
        } else {
            throw new IOException("can't get channel for service " + message.getService());
        }
        Channel channel = channelFuture.getNow();
        return send(message, future, asyncHandler, channel, requestTimeoutInMs);
    }

    @Override
    public boolean isClose() {
        return status == Status.CLOSED;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        CicadaSender that = (CicadaSender) o;
        return status == that.status &&
                Objects.equals(ipAddress, that.ipAddress);
    }

//    @Override
//    public int hashCode() {
//        return Objects.hash(status, ipAddress);
//    }

    private <T> ListenableFuture<T> send(Message message, ResponseFuture<T> future, AsyncHandler<T> asyncHandler, Channel channel, long requestTimeoutInMs) throws IOException {

        if (Channels.isChannelValid(channel)) {
            if (logger.isDebugEnabled()) {
                logger.debug("Using cached Channel {}\n for message \n{}\n", channel, message);
            }
            future.setState(ResponseFuture.STATE.POOLED);
            future.attachChannel(channel);
            Channels.setAttribute(channel, future);
            try {
                writeRequest(message, future, channel, requestTimeoutInMs);
            } catch (Exception ex) {
                logger.error("writeRequest failure", ex);
                try {
                    asyncHandler.onThrowable(ex);
                } catch (Throwable t) {
                    logger.warn("doConnect.writeRequest()", t);
                }
                IOException ioe = new IOException(ex.getMessage());
                ioe.initCause(ex);
                throw ioe;
            }

        } else {
            // bad luck, the channel was closed in-between
            // there's a very good chance onClose was already notified but the future wasn't already registered
            handleUnexpectedClosedChannel(channel, future, Util.REMOTELY_CLOSED_EXCEPTION);
        }
        return future;
    }

    private <T> void writeRequest(Message message, ResponseFuture<T> future, Channel channel, long requestTimeoutInMs) {

        try {
            channel.writeAndFlush(message, channel.newProgressivePromise()).addListener((ChannelFutureListener) f -> {
                if (!f.isSuccess()) {
                    future.getAsyncHandler().onThrowable(f.cause());
                }
            });
            //针对异步调用的超时设置
            if (Channels.isChannelValid(channel) && (Boolean) message.getHeader(Constants.MESSAGE_HEADER_ASYN)) {
                scheduleTimeouts(future, requestTimeoutInMs);
            }
//            long begin = (long) message.getHeader(MESSAGE_HEADER_BEGIN);
//            logger.info("spend {}ms,before send ", System.currentTimeMillis() - begin);
        } catch (Throwable t) {
            logger.error("Can't write request", t);
            releaseChannel(channel, future.getPoolKey());
        }
    }

    private void scheduleTimeouts(ResponseFuture<?> responseFuture, long requestTimeoutInMs) {
        responseFuture.touch();
        TimeoutHolder timeoutHolder = new TimeoutHolder();
        if (requestTimeoutInMs != -1) {
            Timeout requestTimeout = newTimeout(new RequestTimeoutTimerTask(responseFuture, this, timeoutHolder, requestTimeoutInMs), requestTimeoutInMs);
            timeoutHolder.requestTimeout = requestTimeout;
        }
        responseFuture.setTimeoutHolder(timeoutHolder);
    }

    public Timeout newTimeout(TimerTask task, long delay) {
        return timer.newTimeout(task, delay, TimeUnit.MILLISECONDS);
    }

    public void abortAndReleaseChannel(Channel channel, ResponseFuture<?> future, Throwable t) {
        releaseChannel(channel, future.getPoolKey());
        abort(future, t);
    }

    public void abort(ResponseFuture<?> future, Throwable t) {
        if (!future.isDone()) {
            logger.debug("Aborting Future {}\n", future);
            logger.debug(t.getMessage(), t);
            future.abort(t);
        }
    }

    public void handleUnexpectedClosedChannel(final Channel channel, final ResponseFuture<?> future, final Throwable t) {

        logger.info("{} closed by remote server ", channel);
        if (!future.isDone()) {
            abortAndReleaseChannel(channel, future, t);
        } else {
            releaseChannel(channel, future.getPoolKey());
        }
    }

    //添加释放链接，针对连接池无法递减连接数
    public void releaseChannel(Channel channel, String poolKey) {
//        Channels.removeAttribute(channel);
        channelPoolManager.getChannelPool(poolKey).release(channel);

    }

    public void releaseChannel(Channel channel) {
//        Channels.removeAttribute(channel);
        channelPoolManager.getPools().forEach(channelPool -> channelPool.release(channel));
    }

    @Override
    public String toString() {
        return "CicadaSender{" +
                "status=" + status +
                ", ipAddress=" + ipAddress +
                '}';
    }
}
